package com.gkoudai.open.platform.client;

import com.gkoudai.open.platform.entry.Property;
import com.gkoudai.open.platform.handler.DataHandler;
import com.gkoudai.open.platform.listener.*;
import io.socket.client.IO;
import io.socket.client.Socket;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import okhttp3.OkHttpClient;
import org.springframework.util.StringUtils;

import java.net.URI;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

import static com.gkoudai.open.platform.common.StringCommon.parseJsonString;
import static com.gkoudai.open.platform.common.SubscribeEnum.*;

/**
 * @author : Amater
 * @date : 2024/10/28 20:15
 */
@Slf4j
public class SocketIoClient {
    private Property property;
    private Socket socket;
    private DataHandler dataHandler;

    public SocketIoClient(Property property, DataHandler dataHandler) {
        this.dataHandler = dataHandler;
        this.property = property;
        OkHttpClient.Builder clientBuilder = new OkHttpClient.Builder()
                .connectTimeout(property.getSocketConnectTimeoutMills(), TimeUnit.MILLISECONDS)
                .readTimeout(property.getSocketReadTimeoutMills(), TimeUnit.MILLISECONDS)
                .writeTimeout(property.getSocketWriteTimeoutMills(), TimeUnit.MILLISECONDS);
        IO.Options options = new IO.Options();
        options.transports = new String[]{"websocket"};
        options.reconnection = false;
        options.callFactory = clientBuilder.build();
        options.query = "accessToken=" + property.getToken();
        URI uri = URI.create("https://open-socket.gkoudai.com");
        socket = IO.socket(uri, options);
        socket.on(Socket.EVENT_CONNECT, new ConnectListener(this, dataHandler));
        socket.on(Socket.EVENT_DISCONNECT, new DisconnectListener(dataHandler, this));
        socket.on(Socket.EVENT_CONNECT_ERROR, new ConnectErrorListener(dataHandler, this));
        socket.on(SUBSCRIBE_QUOTE.getEventName(), new SubscribeQuoteResponseListener(dataHandler));
        socket.on(SUBSCRIBE_STOCK_QUOTE.getEventName(), new SubscribeStockQuoteResponseListener(dataHandler));
        socket.on(QUOTE_RESPONSE.getEventName(), new QuotesResponseListener(dataHandler));
        socket.on(SUBSCRIBE_NEWS.getEventName(), new SubscribeNewsResponseListener(dataHandler));
        socket.on(NEWS_RESPONSE.getEventName(), new NewsResponseListener(dataHandler));
        socket.on(NEW_FUTURES_QUOTES_RESPONSE.getEventName(), new NewFuturesQuoteDataListener(dataHandler));
        socket.on(SUBSCRIBE_NEW_FUTURES_QUOTES.getEventName(), new SubscribeNewFuturesQuoteListener(dataHandler));
        socket.connect();
    }

    public void reconnect() {
        //此处限制重连频率
        waitTime();
        log.info("socket reconnect");
        if (dataHandler != null) {
            if (!dataHandler.beforeReconnect(this)) {
                return;
            }
        }
        socket.close();
        OkHttpClient.Builder clientBuilder = new OkHttpClient.Builder()
                .connectTimeout(property.getSocketConnectTimeoutMills(), TimeUnit.MILLISECONDS)
                .readTimeout(property.getSocketReadTimeoutMills(), TimeUnit.MILLISECONDS)
                .writeTimeout(property.getSocketWriteTimeoutMills(), TimeUnit.MILLISECONDS);
        IO.Options options = new IO.Options();
        options.transports = new String[]{"websocket"};
        options.reconnection = false;
        options.callFactory = clientBuilder.build();
        options.query = "accessToken=" + property.getToken();
        URI uri = URI.create("https://open-socket.gkoudai.com");
        socket = IO.socket(uri, options);
        socket.on(Socket.EVENT_CONNECT, new ConnectListener(this, dataHandler));
        socket.on(Socket.EVENT_DISCONNECT, new DisconnectListener(dataHandler, this));
        socket.on(Socket.EVENT_CONNECT_ERROR, new ConnectErrorListener(dataHandler, this));
        socket.on(SUBSCRIBE_QUOTE.getEventName(), new SubscribeQuoteResponseListener(dataHandler));
        socket.on(SUBSCRIBE_STOCK_QUOTE.getEventName(), new SubscribeStockQuoteResponseListener(dataHandler));
        socket.on(QUOTE_RESPONSE.getEventName(), new QuotesResponseListener(dataHandler));
        socket.on(SUBSCRIBE_NEWS.getEventName(), new SubscribeNewsResponseListener(dataHandler));
        socket.on(NEWS_RESPONSE.getEventName(), new NewsResponseListener(dataHandler));
        socket.on(NEW_FUTURES_QUOTES_RESPONSE.getEventName(), new NewFuturesQuoteDataListener(dataHandler));
        socket.on(SUBSCRIBE_NEW_FUTURES_QUOTES.getEventName(), new SubscribeNewFuturesQuoteListener(dataHandler));
        socket.connect();
        if (Boolean.TRUE.equals(property.getEnableQuote())) {
            subscribeQuote();
            subscribeStockQuote();
        }
        if (Boolean.TRUE.equals(property.getEnableNews())) {
            subscribeNews();
        }
        if (Boolean.TRUE.equals(property.getEnableFuturesQuote())) {
            subscribeNewFuturesQuote();
        }

        if (dataHandler != null) {
            dataHandler.afterReconnect(this);
        }

    }

    @SneakyThrows
    private void waitTime() {
        log.info("即将在{}秒后进行重连",property.getSocketReconnectionIntervalMills()/1000);
        Thread.sleep(property.getSocketReconnectionIntervalMills());
    }

    public void subscribeQuote() {
        if (Objects.isNull(property) || StringUtils.isEmpty(property.getQuoteParams())) {
            return;
        }
        Map<String, String> subQuote = new HashMap<>();
        subQuote.put("accessToken", property.getToken());
        subQuote.put("quoteParam", String.join(",", property.getQuoteParams()));
        socket.emit(SUBSCRIBE_QUOTE.getEventName(), parseJsonString(subQuote));
    }

    public void subscribeNewFuturesQuote() {
        if (Objects.isNull(property) || StringUtils.isEmpty(property.getFuturesQuoteParams())) {
            return;
        }
        Map<String, String> subQuote = new HashMap<>();
        subQuote.put("accessToken", property.getToken());
        subQuote.put("quoteParam", String.join(",", property.getFuturesQuoteParams()));
        socket.emit(SUBSCRIBE_NEW_FUTURES_QUOTES.getEventName(), parseJsonString(subQuote));
    }


    public void subscribeStockQuote() {
        if (Objects.isNull(property) || StringUtils.isEmpty(property.getStockQuoteParams())) {
            return;
        }
        Map<String, String> subQuote = new HashMap<>();
        subQuote.put("accessToken", property.getToken());
        subQuote.put("quoteParam", String.join(",", property.getStockQuoteParams()));
        socket.emit("subscribeStockQuote", parseJsonString(subQuote));
    }

    public void subscribeNews() {
        if (Objects.isNull(property) || StringUtils.isEmpty(property.getMark())) {
            return;
        }
        Map<String, String> subQuote = new HashMap<>();
        subQuote.put("accessToken", property.getToken());
        subQuote.put("mark", String.join(",", property.getMark()));
        socket.emit("subscribeNews", parseJsonString(subQuote));
    }

    public Socket getSocket() {
        return socket;
    }

}
